//
//  RxViewModel.swift
//  live
//
//  Created by melvyn on 2018/9/19.
//  Copyright © 2018 NEET. All rights reserved.
//

import Foundation
import RxSwift

protocol RxViewModel: class {
    
    var isActive: Bool { get set }
    var activeObservable: BehaviorSubject<Bool> { get }
    var didBecomeActive: Observable<Self> { get }
    var didBecomeInActive: Observable<Self> { get }
    func forwardSignalWhileActive<T>(_ observable: Observable<T>) -> Observable<T>
    func throttleSignalWhileInactive<T>(_ observable: Observable<T>) -> Observable<T>
}

private struct RxVMAssociatedKeys {
    static var activeKey = "RxViewModel_active_Key"
    static var observableKey = "RxViewModel_observable_Key"
}

extension RxViewModel {
    
    /// Public «active» variable
    var isActive: Bool {
        get { return (objc_getAssociatedObject(self, &RxVMAssociatedKeys.activeKey) as? Bool) ?? false }
        set {

            if newValue == isActive { return }
            
            objc_setAssociatedObject(self, &RxVMAssociatedKeys.activeKey, newValue, .OBJC_ASSOCIATION_ASSIGN)
            activeObservable.on(.next(newValue))
        }
    }
    
    var activeObservable: BehaviorSubject<Bool> {
        get {
            guard let subj = objc_getAssociatedObject(self, &RxVMAssociatedKeys.observableKey) as? BehaviorSubject<Bool> else {
                
                let subj = BehaviorSubject.init(value: isActive)
                objc_setAssociatedObject(self, &RxVMAssociatedKeys.observableKey, subj, .OBJC_ASSOCIATION_RETAIN_NONATOMIC)
                return subj
            }
            return subj
        }
    }
    
    /**
     Rx `Observable` for the `active` flag. (when it becomes `true`).
     
     Will send messages only to *new* & *different* values.
     */
    var didBecomeActive: Observable<Self> {
        return activeObservable.filter { $0 == true }
            .map { [unowned self] _ in self }
    }
    
    /**
     Rx `Observable` for the `active` flag. (when it becomes `false`).
     
     Will send messages only to *new* & *different* values.
     */
    var didBecomeInActive: Observable<Self> {
        return activeObservable.filter { $0 == false }
            .map { [unowned self] _ in self }
    }
    
    /**
     Subscribes (or resubscribes) to the given signal whenever
     `didBecomeActiveSignal` fires.
     
     When `didBecomeInactiveSignal` fires, any active subscription to `signal` is
     disposed.
     
     Returns a signal which forwards `next`s from the latest subscription to
     `signal`, and completes when the receiver is deallocated. If `signal` sends
     an error at any point, the returned signal will error out as well.
     */
    func forwardSignalWhileActive<T>(_ observable: Observable<T>) -> Observable<T> {
        let signal = activeObservable
        
        return Observable.create { (obs: AnyObserver<T>) -> Disposable in
            let disposable = CompositeDisposable()
            var signalDisposable: Disposable? = nil
            var disposeKey: CompositeDisposable.DisposeKey?
            
            let activeDisposable = signal.subscribe( onNext: { active in
                if active == true {
                    signalDisposable = observable.subscribe( onNext: { value in
                        obs.on(.next(value))
                    }, onError: { error in
                        obs.on(.error(error))
                    }, onCompleted: {})
                    
                    if let sd = signalDisposable { disposeKey = disposable.insert(sd) }
                } else {
                    if let sd = signalDisposable {
                        sd.dispose()
                        if let dk = disposeKey {
                            disposable.remove(for: dk)
                        }
                    }
                }
            }, onError: { error in
                obs.on(.error(error))
            }, onCompleted: {
                obs.on(.completed)
            })
            
            _ = disposable.insert(activeDisposable)
            
            return disposable
        }
    }
    
    /**
     Throttles events on the given `observable` while the receiver is inactive.
     
     Unlike `forwardSignalWhileActive:`, this method will stay subscribed to
     `observable` the entire time, except that its events will be throttled when the
     receiver becomes inactive.
     
     - parameter observable: The `Observable` to which this method will stay
     subscribed the entire time.
     
     - returns: Returns an `observable` which forwards events from `observable` (throttled while the
     receiver is inactive), and completes when `observable` completes or the receiver
     is deallocated.
     */
    public func throttleSignalWhileInactive<T>(_ observable: Observable<T>) -> Observable<T> {
        let result = ReplaySubject<T>.create(bufferSize: 1)
        let throttleTime: TimeInterval = 2
        let activeSignal = self.activeObservable.takeUntil(Observable.create { (_: AnyObserver<T>) -> Disposable in
            observable.subscribe(onCompleted: {
                defer { result.dispose() }
                
                result.on(.completed)
            })
        })
        
        _ = Observable.combineLatest(activeSignal, observable) { (active, obs) -> (Bool?, T) in (active, obs) }
            .throttle(throttleTime) { (active: Bool?, _: T) -> Bool in
                return active == false
            }.subscribe(onNext: { (value: (Bool?, T)) in
                result.on(.next(value.1))
            }, onError: { _ in }, onCompleted: {
                result.on(.completed)
            })
        
        return result
    }
}

/**
 Throttles `next`s for which `predicate` returns `true`.
 
 When `valuesPassingTest` returns `true` for a `next`:
 
 1. If another `next` is received before `interval` seconds have passed, the
 prior value is discarded. This happens regardless of whether the new
 value will be throttled.
 2. After `interval` seconds have passed since the value was originally
 received, it will be forwarded on the scheduler that it was received
 upon.
 
 When `valuesPassingTest` returns NO for a `next`, it is forwarded immediately,
 without any throttling.
 
 - parameter interval: The number of seconds for which to buffer the latest value that
 passes `valuesPassingTest`.
 - parameter valuesPassingTest: Passed each `next` from the receiver, this closuer returns
 whether the given value should be throttled.
 
 - returns: Returns a signal which sends `next` events, throttled when `predicate`
 returns `true`. Completion and errors are always forwarded immediately.
 */
extension ObservableType {
    public func throttle(_ interval: TimeInterval, valuesPassingTest predicate: @escaping (E) -> Bool) -> Observable<E> {
        return Observable.create { (obs: AnyObserver<E>) -> Disposable in
            let disposable = CompositeDisposable()
            let scheduler = ConcurrentDispatchQueueScheduler(qos: .default)
            let nextDisposable = SerialDisposable()
            var hasNextValue = false
            var nextValue: E?
            let parent = self.asObservable()
            
            let subscriptionDisposable = parent.subscribe(onNext: {
                /**
                 Disposes the «last» `next` subscription if there was a previous value it gets
                 flushed to the observable `o`.
                 
                 - parameter send:     `Bool` flag indicating where or not the `next` value should be
                 «flushed» to the `observable` `o` or not.
                 */
                func flushNext(_ send: Bool) {
                    nextDisposable.dispose()
                    
                    guard let nV = nextValue, hasNextValue, send else { return }
                    
                    nextValue = nil
                    hasNextValue = false
                    
                    obs.on(.next(nV))
                }
                
                let shouldThrottle = predicate($0)
                flushNext(false)
                
                if !shouldThrottle {
                    obs.on(.next($0))
                    
                    return
                }
                
                hasNextValue = true
                nextValue = $0
                
                let flush = flushNext
                let d = Observable<Int64>.timer(interval, scheduler: scheduler)
                    .subscribe(onNext: { _ in
                        flush(true)
                    })
                
                _ = disposable.insert(d)
            })
            
            _ = disposable.insert(subscriptionDisposable)
            
            return disposable
        }
    }
}
